

package etcdserver

import (
	"expvar"
	"fmt"
	"go.etcd.io/etcd/raft"
	"log"
	"sync"
	"time"

	"go.etcd.io/etcd/client/pkg/v3/logutil"
	"go.etcd.io/etcd/pkg/v3/contention"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
	serverstorage "go.etcd.io/etcd/server/v3/storage"
	"go.uber.org/zap"
)

const (
	// etcd的最大吞吐量不超过100MB/s（100K*1KB值）。
	// 假设RTT大约为10ms，1MB的最大大小足够大。
	maxSizePerMsg = 1 * 1024 * 1024
	// 切勿使rafthttp缓冲区溢出，该缓冲区为4096。
	// TODO:一个更好的常数？
	maxInflightMsgs = 4096 / 8
)

var (
	// 保护raftStatus 
	raftStatusMu sync.Mutex
	// expvar func接口的间接寻址
	// 发布重复名称
	// expvar不支持删除注册名称
	// 因此只注册调用raftStatus 
	// 的函数，并将raftStatus更改为我们需要。
	raftStatus func() raft.Status
)

func init() {
	expvar.Publish("raft.status", expvar.Func(func() interface{} {
		raftStatusMu.Lock()
		defer raftStatusMu.Unlock()
		if raftStatus == nil {
			return nil
		}
		return raftStatus()
	}))
}

// 应用包含要应用的条目、快照。一旦
// 一个应用程序被使用，条目将被同时持久化到
// 以进行raft存储；应用程序必须先读取
// raft DONE，然后才能假定raft消息是稳定的。
type apply struct {
	entries  []raftpb.Entry
	snapshot raftpb.Snapshot
	// notifyc同步etcd服务器应用于raft节点
	notifyc chan struct{}
}

// 结构体 ra位Node 中内嵌了 ra丘NodeConfig ，而在结构体raftNodeConfig 中又内嵌了 etcd-raft 模块中的 Node 。
// 这样 raftNode 实例就可以与 etcd-raft模块完成交互。
type raftNode struct {
	lg *zap.Logger
	tickMu *sync.Mutex
	raftNodeConfig
	// etcd-raft 模块通过返回 Ready 实例与上层模块进行交互，
	// 其中 Ready.Message 字段记录了待发送的消息 ， 其中可能会包含 MsgSnap 类型的消 息，
	// 该类型消息中封装了需要发送到其他节点 的快 照数据。当 raftNode 收到 MsgSnap 消 息之后 ，
	// 会将其写入 msgSnapC 通道中，并等待上层模块进行发送。
	msgSnapC chan raftpb.Message
	// 在 etcd-raft 模块返回的 Ready 实例中， 除了封装了待持久化的 Entry 记录和待持久化的快照数据 ，
	// 还封装了待应用的 Entry 记录。raftNode 会将待应用的记录和快照数据封装成 apply 实例之后写入 applyc 通道等待上层模块处理。
	applyc chan apply
	// Readyc.ReadStates 中封装了只 读请求相关的ReadState 实例，
	// 其中 的最后一项将会被写入 readStateC 通道中 等待上层模块处理。
	readStateC chan raft.ReadState
	// 该定时器就是逻辑时钟， 每触发一次就会推进一次底层的选举计时器和心跳计时器。
	ticker *time.Ticker
	// 检测发往同一节点 的两次心跳消息是否超时，如果超时， 则会打印相关警告信息。
	td *contention.TimeoutDetector
	stopped chan struct{}
	done    chan struct{}
}

type raftNodeConfig struct {
	lg *zap.Logger
	// 该函数用来检测指定节点是否己经被移出当前集群。
	isIDRemoved func(id uint64) bool
	raft.Node
	// raftLog.storage 字段指向的MemoryStorage 为同一实例，主要用来保存持久化的 Entry 记录和快照数据
	raftStorage *raft.MemoryStorage
	// 注意该字段的类型，在 etcd-raft 模块中有一个与之同名的接口（ raft. Storage 接口） ,
	// MemoryStorage 就是 raft.Storage 接口的实现之一。
	storage     serverstorage.Storage
	// 逻辑时钟的刻度
	heartbeat   time.Duration // 用于记录
	// 通过网络将消息发送到 集群中其他节点 ，
	transport rafthttp.Transporter
}

// newRaftNode（）函数，其参数是 raftNodeConfig 实例， 该函数的主要功能就是创建各种通道，以及创建 raftNode 实例，
func newRaftNode(cfg raftNodeConfig) *raftNode {
	var lg raft.Logger
	if cfg.lg != nil {
		lg = NewRaftLoggerZap(cfg.lg)
	} else {
		lcfg := logutil.DefaultZapLoggerConfig
		var err error
		lg, err = NewRaftLogger(&lcfg)
		if err != nil {
			log.Fatalf("cannot create raft logger %v", err)
		}
	}
	raft.SetLogger(lg)
	r := &raftNode{
		lg:             cfg.lg,
		tickMu:         new(sync.Mutex),
		raftNodeConfig: cfg,
		// 为raft心跳消息设置争用检测器。
		// 希望在2次心跳间隔内发送心跳。
		td:         contention.NewTimeoutDetector(2 * cfg.heartbeat),
		readStateC: make(chan raft.ReadState, 1),
		msgSnapC:   make(chan raftpb.Message, maxInFlightMsgSnap),	// 注意该通道的缓冲区大小
		applyc:     make(chan apply),
		stopped:    make(chan struct{}),
		done:       make(chan struct{}),
	}
	if r.heartbeat == 0 {	// 根据 raftNodeConfig.hearbeat 字段创建逻辑时钟，其时间刻度是 heartbeat
		r.ticker = &time.Ticker{}
	} else {
		r.ticker = time.NewTicker(r.heartbeat)	// 每隔 heartbeat 触发一 次
	}
	return r
}

// 木筏。节点在Raft包
func (r *raftNode) tick() {
	r.tickMu.Lock()
	r.Tick()
	r.tickMu.Unlock()
}

// 完成 raftNode 实例的创建之后 ，就可以通过 raftNode.start（）方法启动相关服务，
// 在该方法中会启动一个独立的后台 goroutine ，
// 在该后台 goroutine 中完成了绝大部分与底层 etcd-raft 模块交互的功能
func (r *raftNode) start(rh *raftReadyHandler) {
	internalTimeout := time.Second

	go func() {	// 启动后台的 goroutine 提供服务
		defer r.onStop()
		islead := false	// 刚启动时会将当前节点标识为 follower

		for {
			select {
			case <-r.ticker.C:	// 计时器到期被触发，调用 Tick （）方法才住进选举计时器和心跳计时器
				r.tick()
			case rd := <-r.Ready():

				// raftNode.start（）方法对etcd-raft 模块返回的 Ready 实例中的 SoftState 字段，以及 ReadStates 字段的处理过程
				// 在 SoftState 中主要封装了当前集群的 Leader 信息和当前节点角色
				if rd.SoftState != nil {
					// 检测集群的 Leader 节点是否发生变化，并记录相关监控信息 ，
					newLeader := rd.SoftState.Lead != raft.None && rh.getLead() != rd.SoftState.Lead
					if newLeader {
						leaderChanges.Inc()
					}

					if rd.SoftState.Lead == raft.None {
						hasLeader.Set(0)
					} else {
						hasLeader.Set(1)
					}
					// 更新 raftNode.lead 字段，将其更新为新的 Leader 节点 ID,
					rh.updateLead(rd.SoftState.Lead)
					// 记录当前节点是否为 Leader 节点
					islead = rd.RaftState == raft.StateLeader
					if islead {
						isLeader.Set(1)
					} else {
						isLeader.Set(0)
					}
					// 调用 raftReadyHandler 中的 updateLeadership（）回调方法，其中会根据 Leader 节点是否发生变化，完成一些操作
					rh.updateLeadership(newLeader)
					r.td.Reset()	// 重置全部探测器中的全部记录
				}

				if len(rd.ReadStates) != 0 {
					select {
					// 将 Ready.ReadStates 中的最后一项写入 readStateC 通道中
					case r.readStateC <- rd.ReadStates[len(rd.ReadStates)-1]:
					case <-time.After(internalTimeout):
						// 如果上层应用 一直没有读取写入 readStateC 远远中 的 ReadState 实例，会导致本次写入阻塞，
						// 这里会等待 ls ，如果依然无法写入，则放弃写入并输出警告日志
						r.lg.Warn("timed out sending read state", zap.Duration("timeout", internalTimeout))
					case <-r.stopped:
						return
					}
				}

				// raftNode.start （） 方法对 Ready 实例中待应用的 Entry 记录，以及快照数据的处理流程
				notifyc := make(chan struct{}, 1)	// 创建 notifyc 通道
				// 将 Ready 实例中的待应用 Entry 记录以及快照数据封装成 apply 实例 ，其中封装了 notifyc 通道， 该通
				// 道用来协调当前 goroutine 和 EtcdServer 启动的后台 goroutine 的执行
				ap := apply{
					entries:  rd.CommittedEntries,	// 已提交、待应用的 Entry 记录
					snapshot: rd.Snapshot,	// 待持久化的快照数据
					notifyc:  notifyc,
				}
				// 更新 EtcdServer 中记录的己提交位置（ EtcdServer . committedindex 字段），
				updateCommittedIndex(&ap, rh)

				select {
				case r.applyc <- ap:	// 将apply实例 写入applyc通道中，等待上层应用读取并进行处理
				case <-r.stopped:
					return
				}

				// 如果当前节点处于 Leader 状态，则 raftNode.start（） 方法会先调用 raftNode. processMessages()
				// 方法对待发送的消息进行过滤， 然后调用 rafNode.transport.Send（） 方法完成消息的发送，
				if islead {
					// gofail:var raftBeforeLeaderSend结构{}
					r.transport.Send(r.processMessages(rd.Messages))
				}

				// 必须先保存快照文件和WAL快照条目，然后再将任何其他条目或硬状态保存到
				// 确保快照还原后的恢复是可能的。
				if !raft.IsEmptySnap(rd.Snapshot) {
					// gofail:var raftBeforeSaveSnap结构{}
					if err := r.storage.SaveSnap(rd.Snapshot); err != nil {
						r.lg.Fatal("failed to save Raft snapshot", zap.Error(err))
					}
					// gofail:var raftBeforeSaveSnap结构{}
				}

				// 通过 raftNode.storage 将 Ready 实例中携带的 HardState 信息和待持久化的 Entry 记录写入WAL日志文件中，
				if err := r.storage.Save(rd.HardState, rd.Entries); err != nil {
					r.lg.Fatal("failed to save Raft hard state and entries", zap.Error(err))
				}
				if !raft.IsEmptyHardState(rd.HardState) {
					proposalsCommitted.Set(float64(rd.HardState.Commit))
				}
				// gofail:var raftBeforeSaveSnap结构{}

				if !raft.IsEmptySnap(rd.Snapshot) {
					// 通过 raftNode.storage 将 Ready 实例中携带的快照数据保存到磁盘中
					if err := r.storage.Sync(); err != nil {
						r.lg.Fatal("failed to sync Raft snapshot", zap.Error(err))
					}

					// 在后面介绍的 EtcdServer 中会启动后台 goroutine 读取前面介绍的 a pp lye 遥遥，并处理 apply 中
					// 封装快照数据。 这里使用 notifyc 迢迢通知该后台 go routine ，该 apply 实例中的快照数据已经
					// 被持久化到磁盘 ，后 台 goroutine 可以开始应用该快照数据了
					notifyc <- struct{}{}

					// gofail:var raftBeforeApplySnap struct{}
					r.raftStorage.ApplySnapshot(rd.Snapshot)	// 将快照数据保存到 MemoryStorage 中
					r.lg.Info("applied incoming Raft snapshot", zap.Uint64("snapshot-index", rd.Snapshot.Metadata.Index))
					// gofail:var raftaftereapplysnap struct{}

					if err := r.storage.Release(rd.Snapshot); err != nil {
						r.lg.Fatal("failed to release Raft wal", zap.Error(err))
					}
					// gofail:var raftAfterWALRelease struct{}
				}
				// 将待持久化的 Entry 记录写入 MemoryStorage 中
				r.raftStorage.Append(rd.Entries)

				if !islead {
					// 与 Leader 节点的处理逻辑类似
					msgs := r.processMessages(rd.Messages)

					// 处理 Ready 实例的过程基本结束，这里会通知 EtcdServer 启动的后台 goroutine ，检测是否生成快照
					notifyc <- struct{}{}

					// 候选者或跟随者需要等待所有挂起的配置
					// 在发送之前应用更改信息。
					// 否则，我们可能会错误地计算选票（例如，被删除成员的选票）。
					// 同样，在应用层应用配置更改之前，慢速机器的跟随者raft层可以在其自己的单节点集群上继续成为领导者
					// 。
					// 我们现在只需等待所有待处理的条目被应用。
					// 如果它导致不必要的长阻塞问题，我们可能会在以后对此进行改进。
					waitApply := false
					for _, ent := range rd.CommittedEntries {
						if ent.Type == raftpb.EntryConfChange {
							waitApply = true
							break
						}
					}
					if waitApply {
						// 阻塞，直到“applyAll”调用“applyWait”。触发'
						// 与计划的配置更改作业
						// （假设notifyc的上限为1）
						select {
						case notifyc <- struct{}{}:
						case <-r.stopped:
							return
						}
					}

					// gofail:var raftBeforeFollowerSend struct{}
					r.transport.Send(msgs)	// 发送消息
				} else {
					// 处理 Ready 实例的过程基本结束，这里会通知 EtcdServer 启动的后台 go routine ，检测是否生成快照
					notifyc <- struct{}{}
				}
				// 最后调用 raft.node.Advance（）方法，通知 etcd-raft 模块此次 Ready 实例已经处理完成 ，
				// etcd-raft 模块更新相应信息（例如，己应用 Entry 的最大索引值）之后，可以续返回 Ready 实例
				r.Advance()
			case <-r.stopped:
				return	// 当前节点已停止
			}
		}
	}()
}

func updateCommittedIndex(ap *apply, rh *raftReadyHandler) {
	var ci uint64
	if len(ap.entries) != 0 {
		ci = ap.entries[len(ap.entries)-1].Index
	}
	if ap.snapshot.Metadata.Index > ci {
		ci = ap.snapshot.Metadata.Index
	}
	if ci != 0 {
		rh.updateCommittedIndex(ci)
	}
}

// raftNode . processMessages() 方法处理待发送消息的逻辑比较清晰，它首先会对消息进行过滤，
// 去除目标节点己被移出集群的消息，然后分别过滤 MsgAppResp 消息、 MsgSnap 消息和 MsgHeartbeat 消息，
func (r *raftNode) processMessages(ms []raftpb.Message) []raftpb.Message {
	sentAppResp := false
	for i := len(ms) - 1; i >= 0; i-- {	// 从后向前边历全部待发送的消息
		if r.isIDRemoved(ms[i].To) {	// 消息的目标节点已从集群中移除，
			// 将消息的目标节点 ID 设立为 0 ，通过前面对 etcd-raft 模块的分析可知，其发送消息的过程中 ，
			// 会忽略目标节点为 0 的消息
			ms[i].To = 0
		}
		// 只会发送最后一条 MsgAppResp 消息，通过前面对 etcd-raft 模块的分析可知，没有必妥同时发送多条 MsgAppResp 消息
		if ms[i].Type == raftpb.MsgAppResp {
			if sentAppResp {
				ms[i].To = 0
			} else {
				sentAppResp = true
			}
		}

		if ms[i].Type == raftpb.MsgSnap {	// 对 MsgSnap 消息的处理
			select {
			case r.msgSnapC <- ms[i]:	// 将 MsgSnap 消息写入 msgSnapC 通道中
			default:	// 如果msgSηape 通道的缓冲区满了 ， 则放弃此次快照的发送
			}
			ms[i].To = 0	// 将目标节点设豆为 0 ，则 rafNode.transport 后续不会发送该消息
		}
		if ms[i].Type == raftpb.MsgHeartbeat {	// 对 MsgHeartbeat 类型的消息处理
			// 通过 TimeoutDetector 进行检测 ，检测发往目标节点的心跳消息间 隔是否过大
			ok, exceed := r.td.Observe(ms[i].To)
			if !ok {
				// 输出警告日志，表示当前节点可能已经过栽了
				// TODO:限制请求速率。
				r.lg.Warn(
					"leader failed to send out heartbeat on time; took too long, leader is overloaded likely from slow disk",
					zap.String("to", fmt.Sprintf("%x", ms[i].To)),
					zap.Duration("heartbeat-interval", r.heartbeat),
					zap.Duration("expected-duration", 2*r.heartbeat),
					zap.Duration("exceeded-duration", exceed),
				)
				heartbeatSendFailures.Inc()
			}
		}
	}
	return ms
}

func (r *raftNode) apply() chan apply {
	return r.applyc
}

func (r *raftNode) stop() {
	r.stopped <- struct{}{}
	<-r.done
}

func (r *raftNode) onStop() {
	r.Stop()
	r.ticker.Stop()
	r.transport.Stop()
	if err := r.storage.Close(); err != nil {
		r.lg.Panic("failed to close Raft storage", zap.Error(err))
	}
	close(r.done)
}

// 用于测试
func (r *raftNode) pauseSending() {
	p := r.transport.(rafthttp.Pausable)
	p.Pause()
}

func (r *raftNode) resumeSending() {
	p := r.transport.(rafthttp.Pausable)
	p.Resume()
}

// 推进筏节点的节拍。
// 这可用于快进选举
// 在多数据中心部署中使用滴答声，从而加快选举过程。
func (r *raftNode) advanceTicks(ticks int) {
	for i := 0; i < ticks; i++ {
		r.tick()
	}
}
